package cn.smileyan.demos;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 将实体类序列化并写入 kafka
 * @author smileyan
 */
public class KafkaArgsSinkDemo {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        final KafkaArgsBuilder kafkaArgsBuilder = new KafkaArgsBuilder(args);

        DataStream<Student> dataStream = env.fromElements(
                new Student(1, "张三", 18),
                new Student(2, "李四", 20),
                new Student(3, "王五", 22)
        );

        KafkaSink<Student> sinker = kafkaArgsBuilder.buildSink(Student.class);

        dataStream.sinkTo(sinker);

        env.execute();
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    static class Student {
        private Integer id;
        private String name;
        private Integer age;
    }
}
